/*
package com.seaboxdata.threadHandler;

import com.seaboxdata.entity.RealTimeBusCash;
import com.seaboxdata.kafka.BaseStringKafkaConsumer;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import com.seaboxdata.service.RealTimeBusCashService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

*/
/**
 * @author
 * @create 2020-09-10 17:38
 **//*


@Component
@Slf4j
public class RealTimeBusCashHandler implements DisposableBean, Runnable {

    private Thread thread;

    private boolean flag;

    private static final String utcPattern = "yyyy-MM-dd HH:mm:ss";

    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat(utcPattern);

    @Value("${realTimeBusCash.kafka.realTimeBusCashTopic}")
    private String topic;

    @Autowired
    private BaseStringKafkaConsumer baseStringKafkaConsumer;

    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    @Autowired
    private RealTimeBusCashService realTimeBusCashService;


    @Autowired
    public void initThread() {
        this.flag = true;
        this.thread = new Thread(this);
        thread.setName("RealTimeBusCashHandler");
        //this.thread.start();
        log.info("RealTimeBusCashHandler start");
    }


    @Override
    public void run() {
        KafkaConsumer<String, String> consumer = baseStringKafkaConsumer.getConsumer(topic);
        while (flag) {
            try {
                boolean success = true;
                ConsumerRecords<String, String> messages = consumer.poll(Duration.ofMillis(100L));
                List<RealTimeBusCash> busCashs = new ArrayList<>();
                for (ConsumerRecord<String, String> message : messages) {
                    String[] value = message.value().split(",");
                    RealTimeBusCash cash = new RealTimeBusCash();
                    cash.setCardNumber(value[0]);
                    cash.setEquipmentCode(value[1]);
                    try {
                        cash.setCashtime(simpleDateFormat.parse(value[2]));
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("we have cashTime invalid data,cardNumber is{},invalid value is {}", cash.getCardNumber(), value[2]);
                    }
                    cash.setCashAmount(Double.valueOf(value[3]) / 100);
                    cash.setFiled1(value[4]);
                    cash.setFiled2(value[5]);
                    cash.setFiled3(value[6]);
                    cash.setTradeType(Integer.valueOf(value[7]));
                    cash.setCjsj(new Date());
                    busCashs.add(cash);
                }
                if (!CollectionUtils.isEmpty(busCashs)) {
                    try {
                        realTimeBusCashService.saveBatch(busCashs);
                        consumer.commitSync();
                    } catch (Exception e) {
                        log.error("realTimeBusCash commit offset failed");
                        e.printStackTrace();
                        //success = false;
                    }

                    */
/*if (success) {
                        try {
                            realTimeBusCashService.saveBatch(busCashs);
                        } catch (Exception e) {
                            log.error("batch insert realTimeBusCash failed");
                            e.printStackTrace();
                            for (RealTimeBusCash realTimeBusCash : busCashs) {
                                try {
                                    baseStringKafkaProducer.sendMessageToKafka(JSONObject.toJSONString(realTimeBusCash), topic);
                                } catch (Exception e1) {
                                    log.error("write database failed and write kafka realTimeBusCash error");
                                    e1.printStackTrace();
                                }
                            }
                        }
                    }*//*

                }
                busCashs.clear();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("consumer thread have error");
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        this.flag = false;
    }
}
*/
